using System;
using System.Collections.Generic;
using System.Text;
using System.Resources;
using System.Threading;
using System.Diagnostics;
using System.Net;
using System.Net.Sockets;

namespace Microsoft.Samples.BizTalk.Adapter.Tcp
{
    /// <summary>
    ///     This class accepts a pending socket and adds it to the least busy worker thread queue.
    /// </summary>
    /// <author>
    ///     Paul Brown
    /// </author>
    public sealed class Dispatcher : IDisposable
    {
        #region Private Constants

        private const int CheckIdleThreadsPeriod = 5000;    // Check for idle threads every 5 seconds

        #endregion

        #region Private Instance Members

        private bool                                _isDisposed;            // A flag that determines whether the object has been disposed of
        private bool                                _isStopping;            // Determines whether the dispatcher is stopping
        private bool                                _isStopped;             // Determines whether the dispatcher is currently stopped.
        private IAdapter                            _adapter;               // A reference to the adapter
        private List<IWorkerThread>                 _workerThreads;         // An array of the worker threads
        private object                              _workerThreadLock;      // A lock for the array
        private Queue<IWorkItem>                    _workItems;             // A queue of work items to dispatch to worker threads
        private object                              _workItemLock;          // A lock for the queue
        private Thread                              _dispatcherThread;      // The main dispatcher thread
        private ManualResetEvent                    _workSignal;            // This is signalled when work has arrived
        private Timer                               _timer;                 // This timer is used to shut down idle threads
        private int                                 _threadIdleTimeout;     // The time that a thread must be idle before being shut down

        // Delegate used to call Start() method asynchronously from within worker thread
        private delegate void StartMethod();

        #endregion

        #region Constructor

        /// <summary>
        ///     This constructor initializes the instance with an instance of the adapter.
        /// </summary>
        public Dispatcher(IAdapter adapter, int threadIdleTimeout)
        {
            // Validate arguments
            if (adapter == null) throw new ArgumentNullException("adapter");

            // Initialize members
            this._isDisposed = false;
            this._isStopping = false;
            this._isStopped = true;
            this._adapter = adapter;
            this._workerThreads = new List<IWorkerThread>();
            this._workItems = new Queue<IWorkItem>();
            this._workItemLock = new object();
            this._workerThreadLock = new object();
            this._dispatcherThread = null;
            this._workSignal = new ManualResetEvent(false);
            this._timer = null;
            this._threadIdleTimeout = threadIdleTimeout;
        }

        #endregion

		#region Finalizer

		/// <summary>
		///     This method is the finalizer for the class.  It uses the same syntax as
		///     a C++ destructor but is not deterministically destroyed.  By using this
		///     syntax, the compiler automatically emits a try-finally block around the
		///     code to ensure the base class is disposed of as well.
		/// </summary>
		~Dispatcher()
		{
			// Call private implementation of Dispose
			this.Dispose(false);
		}

		#endregion

        #region Public Instance Properties

        /// <summary>
        ///     This property returns whether the object has been disposed of.
        /// </summary>
        public bool IsDisposed
        {
            get { return (this._isDisposed); }
        }

        /// <summary>
        ///     This property returns a reference to the adapter.
        /// </summary>
        public IAdapter Adapter
        {
            get { return (this._adapter); }
        }

        /// <summary>
        ///     This property returns whether the dispatcher is currently stopping.
        /// </summary>
        public bool IsStopping
        {
            get { return (this._isStopping); }
        }

        /// <summary>
        ///     This property returns whether the dispatcher is currently in a stopped state.
        /// </summary>
        public bool IsStopped
        {
            get { return (this._isStopped); }
        }

        /// <summary>
        ///     This property returns the number of minutes a thread must be idle before being shut down.
        /// </summary>
        public int ThreadIdleTimeout
        {
            get { return (this._threadIdleTimeout); }
        }

        #endregion

        #region Public Instance Methods

        /// <summary>
        ///     This method starts the dispatcher.  The argument defines the type and number of each worker thread to start
        ///     as well as whether to lazy initialize the threads in the dispatcher (that is, not start them).
        /// </summary>
        public void Start(KeyValuePair<Type, int>[] workerThreads, bool lazyInitialization)
        {
            if (this.Adapter.TraceSwitch.TraceVerbose)
                TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.DispatcherStarting);
            
            // Start all worker threads of each type
            foreach (KeyValuePair<Type, int> kvp in workerThreads)
            {
                // Create the number of threads identified by the value in the array
                for (int i = 0; i < kvp.Value; i++)
                {
                    // Create an instance of the thread
                    IWorkerThread workerThread = (IWorkerThread)Activator.CreateInstance(kvp.Key, this.Adapter);
                    workerThread.ThreadTerminated += new EventHandler(this.WorkerThread_ThreadTerminated);

                    // Start it (only if not lazy starting)
                    if (!lazyInitialization)
                    {
                        workerThread.Start();
                    }

                    // Add it to the collection
                    this._workerThreads.Add(workerThread);
                }
            }

            // Start a new thread for the dispatcher
            this._dispatcherThread = new Thread(new ThreadStart(this.DispatchWork));
            this._dispatcherThread.IsBackground = true;
            this._dispatcherThread.Name = TraceHelper.ResourceManager.GetString(ResourceConstants.DispatcherThreadName);
            this._dispatcherThread.Start();

            // Start timer to check for idle threads
            this._timer = new Timer(new TimerCallback(this.Dispatcher_Timer));
            this._timer.Change(0, Dispatcher.CheckIdleThreadsPeriod);

            // Dispatcher started, set flags
            this._isStopping = false;
            this._isStopped = false;

            if (this.Adapter.TraceSwitch.TraceVerbose)
                TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.DispatcherStarted);
        }

        /// <summary>
        ///     This method stops the dispatcher from processing any more work.
        /// </summary>
        public void Stop()
        {
            if (!this.IsStopped)
            {
                if (this.Adapter.TraceSwitch.TraceVerbose)
                    TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.DispatcherStopping);

                // Asked to stop, set flag
                this._isStopping = true;
                this._isStopped = false;

                // Shut down dispatcher thread
                if (this._dispatcherThread.ThreadState != System.Threading.ThreadState.Aborted &&
                    this._dispatcherThread.ThreadState != System.Threading.ThreadState.Stopped)
                {
                    lock (this._workItemLock)
                    {
                        // Wait for dispatcher thread to finish processing
                        while (this._workItems.Count > 0) ;

                        // Interrupt thread to stop it
                        this._dispatcherThread.Interrupt();
                    }
                }

                // Stop worker threads
                lock (this._workerThreadLock)
                {
                    foreach (IWorkerThread worker in this._workerThreads)
                    {
                        // Stop it
                        worker.Stop();
                    }
                }

                // Dispatcher stopped, set flags
                this._isStopped = true;
                this._isStopping = false;

                if (this.Adapter.TraceSwitch.TraceVerbose)
                    TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.DispatcherStopped);
            }
        }

        /// <summary>
        ///     This method accepts a new work item.
        /// </summary>
        public void AddWorkItem(IWorkItem workItem)
        {
            // Is the dispatcher stopped or stopping?
            if (this.IsStopping) throw new DispatcherStoppingException();
            if (this.IsStopped) throw new DispatcherStoppedException();

            if (this.Adapter.TraceSwitch.TraceVerbose)
                TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.DispatcherAddingWork, workItem.GetType().Name);

            // Add to queue
            lock (this._workItemLock)
            {
                this._workItems.Enqueue(workItem);
            }

            // Signal dispatcher thread
            this._workSignal.Set();
        }

        #endregion

        #region Private Instance Methods

        /// <summary>
        ///     This method provides a private implementation which a flag that specifies whether we
        ///     are manually disposing of the object, or is the CLR disposing of it for us.
        /// </summary>
        private void Dispose(bool disposing)
        {
            // Only dispose if we are not already
            if (!this.IsDisposed)
            {
                // If we have been GC collected, the CLR may have already disposed of other object references
                // so it may not be possible to clean them up.  If we are manually disposed, we can dispose
                // of our objects.
                if (disposing)
                {
                    // If we are not stopped or stopping, then stop the dispatcher
                    if (this.IsStopping == false && this.IsStopped == false)
                    {
                        this.Stop();
                    }
                }

                // Set flag
                this._isDisposed = true;
            }
        }

        /// <summary>
        ///     This method is the delegate that runs the dispatch thread.
        /// </summary>
        private void DispatchWork()
        {
            try
            {
                // Loop until we are interrupted
                while (true)
                {
                    // Reset wait object
                    this._workSignal.Reset();

                    // Loop while we have work items
                    while (this._workItems.Count > 0)
                    {
                        // Dequeue item
                        IWorkItem workItem = null;
                        lock (this._workItemLock)
                        {
                            workItem = this._workItems.Dequeue();
                        }

                        try
                        {
                            // Check to see what threads can support this work item type and dispatch to least busy worker thread.
                            // Note: This would be the point to implement a different algorithm for dispatching the work to a worker thread.
                            lock (this._workerThreadLock)
                            {
                                // Assign least busy thread
                                IWorkerThread leastBusy = null;

                                foreach (IWorkerThread worker in this._workerThreads)
                                {
                                    // Does this worker thread support this type of work item?
                                    if (worker.WorkItemType == workItem.GetType())
                                    {
                                        // Is the thread stopped or running?  If it is stopping, avoid the thread to allow it to
                                        // continue shutting down before becoming a candidate for reuse.
                                        if (!worker.IsStopping)
                                        {
                                            if (leastBusy == null)
                                            {
                                                // No thread assigned yet, store first one
                                                leastBusy = worker;
                                            }
                                            else
                                            {
                                                // Store least busy thread
                                                if (worker.CurrentWorkLoad < leastBusy.CurrentWorkLoad)
                                                {
                                                    leastBusy = worker;
                                                }
                                            }
                                        }
                                    }
                                }

                                // Dispatch work to thread
                                if (leastBusy != null)
                                {
                                    if (this.Adapter.TraceSwitch.TraceVerbose)
                                        TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.DispatcherWorkDispatchedToThread, workItem.GetType().Name, leastBusy.Name);

                                    // Start thread if not running
                                    if (leastBusy.IsStopped)
                                    {
                                        leastBusy.Start();
                                    }

                                    // Dispatch to thread
                                    leastBusy.DoWork(workItem);
                                }
                                else
                                {
                                    // If no worker threads, something badly wrong or all stopping!
                                    throw new NoWorkerThreadsException();
                                }
                            }
                        }
                        catch (NoWorkerThreadsException e)
                        {
                            Trace.TraceError(e.Message);

                            // Bad error, should never or rarely happen (there is the potential to lose work here if it does though!)
                            this.Adapter.RestartAdapter(e);
                        }
                        catch (Exception e)
                        {
                            TraceHelper.WriteEntry(TraceLevel.Error, ResourceConstants.DispatcherGeneralWorkAssignmentError, e.Message);

                            // Enqueue item again
                            if (workItem != null)
                            {
                                lock (this._workItemLock)
                                {
                                    this._workItems.Enqueue(workItem);
                                }
                            }
                        }
                    }

                    if (this.Adapter.TraceSwitch.TraceVerbose)
                        TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.DispatcherWaitingForWork);

                    // Wait for signal that work has arrived
                    this._workSignal.WaitOne();
                }
            }
            catch (ThreadInterruptedException)
            {
                // Asked to shut down
                if (this.Adapter.TraceSwitch.TraceVerbose)
                    TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.DispatcherThreadInterrupted);
            }
            catch (Exception e)
            {
                TraceHelper.WriteEntry(TraceLevel.Error, ResourceConstants.DispatcherGeneralError, e.Message);

                // Bad error, restart adapter
                this.Adapter.RestartAdapter(e);
            }
        }

        #endregion

        #region Private Instance Event Handlers

        /// <summary>
        ///     This handler is invoked whenever the timer ticks.
        /// </summary>
        private void Dispatcher_Timer(object state)
        {
            try
            {
                if (this._workerThreads != null)
                {
                    // Lock array as we don't want anyone else enumerating and assigning work while we maybe shutting a
                    // thread down.
                    lock (this._workerThreadLock)
                    {
                        // Loop through worker threads
                        foreach (IWorkerThread worker in this._workerThreads)
                        {
                            // Check to see if the worker thread is idle and how long it has been idle for
                            if (worker.IsStopping == false && worker.IsStopped == false)
                            {
                                if (worker.IsIdle && worker.IdleTime.Minutes >= this.ThreadIdleTimeout)
                                {
                                    if (this.Adapter.TraceSwitch.TraceVerbose)
                                        TraceHelper.WriteEntry(TraceLevel.Verbose, ResourceConstants.ShuttingDownIdleThread, worker.Name);

                                    // Thread has been idling too long, shut it down
                                    worker.Stop(true);
                                }
                            }
                        }
                    }
                }
            }
            catch (Exception e)
            {
                TraceHelper.WriteEntry(TraceLevel.Error, ResourceConstants.FailedCheckingIdleThreads, e.Message);
            }
        }

        /// <summary>
        ///     This event handler is called whenever the worker thread has terminated unexpectedly without
        ///     first calling Stop() in this class.
        /// </summary>
        private void WorkerThread_ThreadTerminated(object sender, EventArgs e)
        {
            // Only invoke Start() method only if we are not stopping/stopped or the thread has idle terminated
            // as this indicates the thread has terminated unexpectedly and should be started again.
            WorkerThread workerThread = (WorkerThread)sender;
            if ((!this.IsStopping) && (!this.IsStopped) && (!workerThread.HasIdleTerminated))
            {
                // Invoke Start() on worker thread asynchronously to avoid running on same thread as the event invocation
                StartMethod startMethod = new StartMethod(workerThread.Start);

                // Invoke method
                IAsyncResult result = startMethod.BeginInvoke(null, null);

                // Wait for it to finish
                result.AsyncWaitHandle.WaitOne();
            }
        }

        #endregion

        #region IDisposable Members

        /// <summary>
        ///     This method disposes of the object manually.
        /// </summary>
        public void Dispose()
        {
            // Dispose
            this.Dispose(true);

            // Suppress finalize as we have already cleaned it up
            GC.SuppressFinalize(this);
        }

        #endregion
    }
}
